package com.atguigu.flink.streamapi.trans;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Created by Smexy on 2023/1/13
 *
 *      在进行每次Map操作时，都需要连接外部的数据库，如何操作?
 *              数据库： Redis
 *                      Mysql
 *
 *      如果在执行某个Function的部分逻辑时，希望在Task(function执行的线程)创建时，只执行一次。
 *      在Task销毁时，只执行一次。
 *          Function 需要 有一对生命周期(出生时，做一件事。死亡时，做一件事)的方法。
 *                  可以通过实现 AbstractRichFunction 获取。
 *
 *          所有的Function都提供了 Rich版本。
 *              MapFunction: 普通版本，青春版。 没有生命周期方法
 *              RichMapFunction: 豪华版本。 既有对应的处理方法，还有生命周期方法！
 */
public class Demo1_Map
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        /*
            map: 1对1
        */
        DataStreamSource<Integer> ds = env.fromElements(1, 2, 3, 4);

        ds.map(new MyMapFunction()).setParallelism(2)
          .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    //RichMapFunction<IN, OUT> extends AbstractRichFunction implements MapFunction<IN, OUT>
    private static class  MyMapFunction extends RichMapFunction<Integer, Integer>
    {

        private String conn;
        //在Task创建(出生)时，只执行一次
        @Override
        public void open(Configuration parameters) throws Exception {
            System.out.println("创建连接...");
            conn = "连接";
        }

        //在Task销毁(死亡)时，只执行一次
        @Override
        public void close() throws Exception {
            System.out.println("释放连接...");
        }

        @Override
        public Integer map(Integer value) throws Exception {

            System.out.println("使用..."+conn);

            return value + 10;
        }
    }
}
